package com.yanduo.report

import com.yanduo.beans.Log
import com.yanduo.utils.RptUtils
import org.apache.commons.lang3.StringUtils
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SparkSession

/**
  * 媒体报表分析
  *  技术点:broadcast 广播变量， extends app 可以不用写main方法
  *  https://www.bilibili.com/video/BV1F4411i7jK?p=23
  *
  * @author Gerry chan
  *  2020/5/3 17:51
  * @version 1.0
  */
object AppAnalyseRpt extends App{
  if (args.length != 3) {
    println(
      """
        |com.yanduo.report.AppAnalyseRpt
        |参数：
        |  输入路径
        |  字典文件路径
        |  输出路径
      """.stripMargin)
    sys.exit()
  }
  val Array(inputPath,dictFilePath, outputPath) = args

  val sparkConf = new SparkConf()
  sparkConf.setAppName(s"${this.getClass.getSimpleName}")
  sparkConf.setMaster("local[*]")

  sparkConf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")

  val sc = new SparkContext(sparkConf)
  val spark = new SparkSession(sc)

  //字典文件, 注意要：collect 动作, 因为是在Driver端定义的，需要让executor获取
  // 在Driver端 toMap 优化执行效率
  private val dictMap: Map[String, String] = sc.textFile(dictFilePath).map(line => {
    val fields = line.split("\t", -1)
    (fields(4), fields(1))
  }).collect().toMap

  // 将字典数据广播executor
  private val broadcast = sc.broadcast(dictMap)

  // 读取原始日志文件
  sc.textFile(inputPath)
    .map(_.split(",", -1))
    .filter(_.length>=85).map(Log(_))
    .filter(log => !log.appid.isEmpty || !log.appname.isEmpty)  // appid 与appname 都是空的过滤掉
    .map(log => {
    var newAppName = log.appname
    if(StringUtils.isNotEmpty(newAppName)) {
      // 当日志中appName 为空的时候，就去广播变量中找，如果广播变量也没有则 给默认值 "未知"
      newAppName = broadcast.value.getOrElse(log.appid, "未知")
    }
    val reqList: List[Double] = RptUtils.caculateReq(log.requestmode, log.requestmode)
    val rtbList: List[Double] = RptUtils.caulateRtb(log.iseffective, log.isbilling, log.isbid, log.adorderid, log.iswin,
      log.winprice, log.adpayment)
    //广告：点击与展示
    val showClickList: List[Double] = RptUtils.caculateShowClick(log.requestmode, log.iseffective)
    // list拼接采用++
    (newAppName, reqList++rtbList++showClickList)
  }).reduceByKey((list1, list2) => {
    list1.zip(list2).map(t=>t._1+t._2)
  }).map(t => t._1+","+t._2.mkString(","))
    .saveAsTextFile(outputPath)

  spark.stop()

}
